#include "Broker.h"

using namespace skelepp::Core::Messaging;

Broker::Broker(void) {
	pthread_mutex_init(&_lock, NULL);
}

Broker::~Broker(void) {
	pthread_mutex_destroy(&_lock);
}
				
void Broker::subscribe(string id, _Subscriber *subscriber) {
	pthread_mutex_lock(&_lock);

	map<string, vector<_Subscriber *>>::iterator subscription = _subscribers.find(id);
	if (subscription == _subscribers.end()) {
		vector<_Subscriber *> subscribers;
		subscribers.push_back(subscriber);
		_subscribers[id] = subscribers;
	} else {
		subscription->second.push_back(subscriber);
	}

	pthread_mutex_unlock(&_lock);
}

void Broker::unsubscribe(string id, _Subscriber *subscriber) {
	pthread_mutex_lock(&_lock);

	map<string, vector<_Subscriber *>>::iterator subscription = _subscribers.find(id);
	if (subscription == _subscribers.end()) {
		// doesn't exist
	} else {
		vector<_Subscriber *>::iterator it = subscription->second.begin();
		for (uint32_t i = 0; it != subscription->second.end(); it++, i++) {
			if (subscription->second.at(i) == subscriber) {
				subscription->second.erase(it);
				break;
			}
		}
	}
	
	pthread_mutex_unlock(&_lock);
}

void Broker::push(_Message *message) {
	pthread_mutex_lock(&_lock);

	map<string, vector<_Subscriber *>>::iterator subscription = _subscribers.find(message->get_id());
	if (subscription == _subscribers.end()) {
		// doesn't exist
	} else {
		for (uint32_t i = 0; i < subscription->second.size(); i++) {
			subscription->second.at(i)->push(message);
		}
	}
	
	pthread_mutex_unlock(&_lock);
}